package com.gin.spark

import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer

object L06_Sample {

  def main(args: Array[String]): Unit = {
    //加载配置 获取spark上下文对象
    val conf = new SparkConf().setMaster("local").setAppName("L01")
    val sc = new SparkContext(conf)
    //只打印错误日志
    sc.setLogLevel("ERROR")

    val data = sc.parallelize(1 to 20, 4)

    //数据抽样
    //1.是否允许数据样本重复  2.数据样本比例(不是精准百分比)  3.随机种子(种子相同,抽取值相同)
    data.sample(true, 0.1, 222).foreach(println)


    println("-------- sample 1 --------")
    println()

    data.sample(false, 0.1, 221).foreach(println)


    println("-------- sample 2 --------")
    println()

    val data2Part = data.mapPartitionsWithIndex(
      (pIdx, pIter) => {
        //pIdx 分区下标
        pIter.map(e => (pIdx, e))
      }
    )

    //大数据分布式移动两种方式: IO移动, shuffle(计算数据归属于哪个分区)
    //重新分区(原始数据块 按计算逻辑分配到 用户指定的分区数上), 可能会触发shuffle(可能涉及到数据从一台机器迁移到另一台)

    //repartition默认true,必定触发shuffle
    //val repartition4 = data2Part.repartition(2)

    //如果分区增加必须有shuffle, 否则无法增加分区; 分区减少可以不用shuffle(一个分区对应多份原始数据了)
    val repartition4 = data2Part.coalesce(2, false)
    val data4Part = repartition4.mapPartitionsWithIndex(
      (pIdx, pIter) => {
        //pIdx 分区下标
        pIter.map(e => (pIdx, e))
      }
    )

    data2Part.foreach(println)
    println("-------- partition test 2 --------")
    println()

    data4Part.foreach(println)
    println("-------- partition test 4 --------")
    println()


  }

}
